package bigdata.backend.datasource.utils;

import bigdata.backend.utils.DbType;

import java.util.*;
import java.util.stream.Collectors;

public class DdlBuilderUtils {

    /**
     * 构建目标数据库的 CREATE TABLE 语句
     *
     * @param columns         字段列表（每个字段为 Map，含 name/type/nullable/default/comment/is_primary_key）
     * @param sourceDbType    源数据库类型
     * @param targetDbType    目标数据库类型
     * @param targetTable     目标表名
     * @param targetSchema    目标 schema（仅 PostgreSQL 使用）
     * @param tableComment    表注释
     * @param includeComments 是否在 SQL 中包含 COMMENT ON 语句（仅 PostgreSQL 同构路径使用）
     * @return 完整的 CREATE TABLE SQL
     */
    public static String buildCreateTableSql(
            List<Map<String, Object>> columns,
            DbType sourceDbType,
            DbType targetDbType,
            String targetTable,
            String targetSchema,
            String tableComment,
            boolean includeComments
    ) {
        // 主键字段必须 NOT NULL（安全兜底）
        for (Map<String, Object> col : columns) {
            if (Boolean.TRUE.equals(col.get("is_primary_key"))) {
                col.put("nullable", false);
            }
        }

        List<String> pkColumns = columns.stream()
                .filter(col -> Boolean.TRUE.equals(col.get("is_primary_key")))
                .map(col -> (String) col.get("name"))
                .collect(Collectors.toList());

        // ========== MySQL / PostgreSQL ==========
        if (targetDbType == DbType.MYSQL || targetDbType == DbType.POSTGRESQL) {
            List<String> columnDefs = new ArrayList<>();
            for (Map<String, Object> col : columns) {
                String def = buildColumnDef(col, sourceDbType, targetDbType, false);
                columnDefs.add(def);
            }

            if (targetDbType == DbType.MYSQL) {
                StringBuilder sql = new StringBuilder();
                sql.append("CREATE TABLE `").append(targetTable).append("` (\n    ")
                        .append(String.join(",\n    ", columnDefs));

                if (!pkColumns.isEmpty()) {
                    String pkClause = pkColumns.stream()
                            .map(name -> "`" + name + "`")
                            .collect(Collectors.joining(", "));
                    sql.append(",\n    PRIMARY KEY (").append(pkClause).append(")");
                }
                sql.append("\n)");

                if (tableComment != null && !tableComment.trim().isEmpty()) {
                    String escaped = tableComment.replace("'", "\\'");
                    sql.append(" COMMENT='").append(escaped).append("'");
                }
                sql.append(";");
                return sql.toString();

            } else { // PostgreSQL
                String schema = Optional.ofNullable(targetSchema).orElse("public");
                StringBuilder sql = new StringBuilder();
                sql.append("CREATE TABLE \"").append(schema).append("\".\"").append(targetTable).append("\" (\n    ")
                        .append(String.join(",\n    ", columnDefs));

                if (!pkColumns.isEmpty()) {
                    String pkClause = pkColumns.stream()
                            .map(name -> "\"" + name + "\"")
                            .collect(Collectors.joining(", "));
                    sql.append(",\n    PRIMARY KEY (").append(pkClause).append(")");
                }
                sql.append("\n);");

                if (includeComments) {
                    List<String> statements = new ArrayList<>();
                    statements.add(sql.toString());

                    // 表注释
                    if (tableComment != null && !tableComment.trim().isEmpty()) {
                        String escaped = tableComment.replace("'", "''");
                        statements.add(String.format(
                                "COMMENT ON TABLE \"%s\".\"%s\" IS '%s';",
                                schema, targetTable, escaped
                        ));
                    }

                    // 字段注释
                    for (Map<String, Object> col : columns) {
                        String comment = (String) col.getOrDefault("comment", "");
                        if (comment != null && !comment.trim().isEmpty()) {
                            String escaped = comment.replace("'", "''");
                            statements.add(String.format(
                                    "COMMENT ON COLUMN \"%s\".\"%s\".\"%s\" IS '%s';",
                                    schema, targetTable, col.get("name"), escaped
                            ));
                        }
                    }
                    return String.join("\n", statements);
                } else {
                    return sql.toString();
                }
            }
        }

        // ========== Doris ==========
        boolean hasExplicitPk = columns.stream().anyMatch(col -> Boolean.TRUE.equals(col.get("is_primary_key")));
        List<Map<String, Object>> keyColumns, valueColumns;
        if (hasExplicitPk) {
            keyColumns = columns.stream().filter(col -> Boolean.TRUE.equals(col.get("is_primary_key"))).collect(Collectors.toList());
            valueColumns = columns.stream().filter(col -> !Boolean.TRUE.equals(col.get("is_primary_key"))).collect(Collectors.toList());
        } else {
            keyColumns = Arrays.asList(columns.get(0));
            valueColumns = columns.subList(1, columns.size());
        }

        List<Map<String, Object>> reordered = new ArrayList<>(keyColumns);
        reordered.addAll(valueColumns);
        List<String> keyNames = keyColumns.stream().map(col -> (String) col.get("name")).collect(Collectors.toList());
        String firstKey = keyNames.get(0);

        List<String> columnDefs = new ArrayList<>();
        for (Map<String, Object> col : reordered) {
            boolean isKey = keyNames.contains(col.get("name"));
            String def = buildColumnDef(col, sourceDbType, DbType.DORIS, isKey);
            columnDefs.add(def);
        }

        StringBuilder sql = new StringBuilder();
        sql.append("CREATE TABLE `").append(targetTable).append("` (\n    ")
                .append(String.join(",\n    ", columnDefs))
                .append("\n)");

        if (hasExplicitPk) {
            String pkClause = keyNames.stream().map(name -> "`" + name + "`").collect(Collectors.joining(", "));
            sql.append("\nUNIQUE KEY(").append(pkClause).append(")");
        } else {
            sql.append("\nDUPLICATE KEY(`").append(firstKey).append("`)");
        }

        if (tableComment != null && !tableComment.trim().isEmpty()) {
            String escaped = tableComment.replace("'", "\\'");
            sql.append(" COMMENT \"").append(escaped).append("\"");
        }

        sql.append("\nDISTRIBUTED BY HASH(`").append(firstKey).append("`) BUCKETS 32\n")
                .append("PROPERTIES(\"replication_num\" = \"1\");");

        return sql.toString();
    }

    /**
     * 映射默认值，确保目标数据库兼容
     */
    private static String mapDefaultValue(String defaultValue, DbType sourceDbType, DbType targetDbType, String columnType) {
        if (defaultValue == null) return null;

        String dv = defaultValue.trim();
        String dvLower = dv.toLowerCase();

        // NULL
        if ("null".equals(dvLower)) {
            return "NULL";
        }

        // 时间函数
        if ("now()".equals(dvLower) || "current_timestamp".equals(dvLower)) {
            if (targetDbType == DbType.MYSQL || targetDbType == DbType.POSTGRESQL || targetDbType == DbType.DORIS) {
                return "CURRENT_TIMESTAMP";
            }
        }

        boolean isStringType = columnType != null &&
                (columnType.toLowerCase().contains("char") ||
                        columnType.toLowerCase().contains("varchar") ||
                        columnType.toLowerCase().contains("string") ||
                        columnType.toLowerCase().contains("text"));

        if (isStringType) {
            if (dv.startsWith("'") && dv.endsWith("'")) {
                return dv;
            } else {
                return "'" + dv + "'";
            }
        }

        // Doris 只支持常量
        if (targetDbType == DbType.DORIS && !isStringType) {
            // 数值类型：只允许纯数字（含小数、负号）
            String clean = dv.replace(".", "").replace("-", "");
            if (clean.chars().allMatch(Character::isDigit)) {
                return dv;
            } else {
                return null; // 丢弃非法默认值
            }
        }

        // 默认：原样返回
        return dv;
    }

    /**
     * 构建单个字段定义
     */
    private static String buildColumnDef(Map<String, Object> col, DbType sourceDbType, DbType targetDbType, boolean isKey) {
        String name = (String) col.get("name");
        String srcType = (String) col.get("type");
        Boolean nullable = (Boolean) col.get("nullable");
        String defaultValue = (String) col.get("default");
        String comment = (String) col.getOrDefault("comment", "");

        // 1. 类型映射
        String mappedType = TypeMapper.mapColumnType(srcType, sourceDbType, targetDbType);

        // 2. 字段名引号
        String colDef;
        if (targetDbType == DbType.MYSQL || targetDbType == DbType.DORIS) {
            colDef = "`" + name + "` " + mappedType;
        } else { // PostgreSQL
            colDef = "\"" + name + "\" " + mappedType;
        }

        // 3. 默认值
        String mappedDefault = mapDefaultValue(defaultValue, sourceDbType, targetDbType, mappedType);
        if (mappedDefault != null) {
            if ("NULL".equals(mappedDefault)) {
                if (targetDbType == DbType.MYSQL || targetDbType == DbType.DORIS) {
                    if (!isKey) {
                        colDef += " DEFAULT NULL";
                    }
                }
            } else {
                colDef += " DEFAULT " + mappedDefault;
            }
        }

        // 4. 非空约束
        if (isKey) {
            colDef += " NOT NULL";
        } else if (nullable != null && !nullable) {
            // Doris 规则：有默认值的非主键字段不能写 NOT NULL
            if (targetDbType == DbType.DORIS && mappedDefault != null) {
                // skip NOT NULL
            } else {
                colDef += " NOT NULL";
            }
        }

        // 5. 内联注释（仅 MySQL/Doris）
        if ((targetDbType == DbType.MYSQL || targetDbType == DbType.DORIS) && comment != null && !comment.trim().isEmpty()) {
            String escaped = comment.replace("'", "\\'");
            colDef += " COMMENT '" + escaped + "'";
        }

        return colDef;
    }
}
